package shardkv

import (
	"sync"
	"time"
)

func (kv *ShardKV) handleDelete(command *ShardCommand, index int) {
	kv.logger.Printf("handleDelete: 开始处理删除命令，分片ID=%d，配置号=%d，Raft索引=%d",
		command.Sid, command.ConfigNum, index)

	reply := ShardReply{}

	//配置已经过期了，或如果不是等待，说明已经删除过了，都可以忽略
	if kv.curConfig.Num != command.ConfigNum {
		kv.logger.Printf("handleDelete: 配置号不匹配（当前=%d, 命令=%d），分片ID=%d，返回ErrDuplicate",
			kv.curConfig.Num, command.ConfigNum, command.Sid)
		reply.Err = ErrDuplicate
	} else if kv.Shards[command.Sid].State != Waiting {
		kv.logger.Printf("handleDelete: 分片状态不是Waiting（当前=%v），分片ID=%d，返回ErrDuplicate",
			kv.Shards[command.Sid].State, command.Sid)
		reply.Err = ErrDuplicate
	} else {
		reply.Err = OK
		//清楚waiting状态，改为ready表示已经删除好了
		kv.Shards[command.Sid].State = Ready
		//清空旧数据
		kv.Shards[command.Sid].DB = make(map[string]string)
		kv.logger.Printf("handleDelete: 成功删除分片数据，分片ID=%d，配置号=%d，状态更新为Ready",
			command.Sid, command.ConfigNum)
	}
	info, exists := kv.pendingDelete[index]
	if exists {
		// 双重校验请求匹配性
		if info.command.Sid == command.Sid && info.command.ConfigNum == command.ConfigNum {
			kv.logger.Printf("handleDelete: 找到匹配的等待请求，Raft索引=%d，发送回复", index)
			go func() {
				info.ch <- &reply
				kv.logger.Printf("handleDelete: 回复已发送至通道，Raft索引=%d", index)
			}()
		} else {
			kv.logger.Printf("handleDelete: 存在等待请求但参数不匹配，Raft索引=%d，忽略发送", index)
		}
	} else {
		kv.logger.Printf("handleDelete: 无匹配的等待请求，Raft索引=%d", index)
	}
}

//上一个配置中分片的持有者收到了notifydelete的通知
func (kv *ShardKV) getDelete(args *ShardArgs, reply *ShardReply) {
	kv.logger.Printf("getDelete: 收到删除通知请求，分片ID=%d，配置号=%d", args.ShardId, args.ConfigNum)
	kv.mu.Lock()
	command := ShardCommand{
		Type:      "getDelete",
		Sid:       args.ShardId,
		ConfigNum: args.ConfigNum,
	}
	index, term, isLeader := kv.rf.Start(command)
	if !isLeader {
		kv.logger.Printf("getDelete: 当前节点不是Leader（分片ID=%d），拒绝处理请求", args.ShardId)
		kv.mu.Unlock()
		reply.Err = ErrWrongLeader
		return
	}
	kv.logger.Printf("getDelete: 命令已提交到Raft，分片ID=%d，Raft索引=%d，任期=%d",
		args.ShardId, index, term)

	//用有缓冲通道来避免出现无人接收的极端情况，减少一个goroutine的浪费
	info := DeleteInfo{
		ch:      make(chan *ShardReply, 1),
		command: &command,
	}
	kv.pendingDelete[index] = &info
	kv.mu.Unlock()

	//最多等待时长
	timer := time.NewTimer(300 * time.Millisecond)
	defer timer.Stop()
	kv.logger.Printf("getDelete: 开始等待处理结果，分片ID=%d，Raft索引=%d", args.ShardId, index)
	for {
		select {
		//在超时之前收到了命令
		case result := <-info.ch:
			kv.clearDelete(index)
			reply.Err, reply.ConfigNum = result.Err, result.ConfigNum
			kv.logger.Printf("getDelete: 收到处理结果，分片ID=%d，错误=%v，配置号=%d",
				args.ShardId, result.Err, result.ConfigNum)
			return
		//等待超时：
		case <-timer.C:
			// 清理映射表
			kv.clearDelete(index)
			//可能当前节点是旧的领导者，处于少数分区，需要重新寻找新的领导者
			reply.Err = ErrWrongLeader
			kv.logger.Printf("getDelete: 等待超时（300ms），分片ID=%d，返回ErrWrongLeader", args.ShardId)
			return
		//没一百毫秒检查一下是否还是领导者且任期没有变化，任期变化即使是领导者，当前命令可能也已经不见了
		case <-time.After(100 * time.Millisecond):
			if kv.rf.StateChanged(term) {
				kv.clearDelete(index)
				reply.Err = ErrWrongLeader
				kv.logger.Printf("getDelete: 任期变化或失去Leader身份，分片ID=%d，返回ErrWrongLeader", args.ShardId)
				return
			}
		}
	}
}

func (kv *ShardKV) checkDelete() {
	for !kv.Killed() {
		if term, isLeader := kv.rf.GetState(); isLeader {
			kv.mu.RLock()
			kv.logger.Printf("checkDelete: 当前为Leader（任期=%d），开始检查需要删除的分片", term)
			startNum := kv.curConfig.Num
			prevConfig := kv.prevConfig
			kv.logger.Printf("checkDelete: 当前配置号=%d，开始处理基于前序配置的分片删除检查", startNum)
			var wg sync.WaitGroup
			//用旧版本是因为需要获取上个分组的地址
			for sid, gid := range prevConfig.Shards {

				//不需要通知
				if kv.Shards[sid].State != Waiting {
					//kv.logger.Printf("notifyShard: 分片ID=%d 状态不是NeedNotify（当前=%v），无需通知", sid, kv.Shards[sid].State)
					continue
				}
				kv.logger.Printf("checkDelete: 分片ID=%d 需要确认删除，目标GID=%d", sid, gid)
				wg.Add(1)
				go func(sid, src, startNum int, addrs []string) {
					defer wg.Done()
					index := 0
					args := ShardArgs{
						Type:      "CheckDelete",
						ShardId:   sid,
						ConfigNum: startNum,
					}
					kv.logger.Printf("checkDelete: 启动确认删除协程，分片ID=%d，目标地址列表=%v", sid, addrs)
					//轮询所有节点找到leader通信
					for {
						//如果节点状态变化过，最好放弃本次拉取，重新拉取
						if kv.rf.StateChanged(term) {
							kv.logger.Printf("notifyShard: 任期变化或不再是Leader，停止确认删除检查ID=%d", sid)
							return
						}
						kv.mu.RLock()
						curNum := kv.curConfig.Num
						kv.mu.RUnlock()
						//配置已经更新了，直接退出，不应该修改其他配置版本的分片状态（之前提交的命令应用成功了）
						if curNum != startNum {
							kv.logger.Printf("checkDelete: 配置号已更新（当前=%d, 开始=%d），停止确认删除检查分片ID=%d",
								curNum, startNum, sid)
							return
						}
						kv.logger.Printf("checkDelete: 尝试通知删除分片ID=%d，（索引=%d）",
							sid, index)
						reply := ShardReply{}
						end := kv.make_end(addrs[index])
						//检查对方是否已经获取分片
						ok := end.Call("ShardKV.HandleShardRequests", &args, &reply)
						//请求成功
						if ok {
							if reply.Err == OK {
								kv.logger.Printf("checkDelete: 对方已进入下一个配置，允许删除分片ID=%d", sid)
								command := ShardCommand{
									Type:      "handleDelete",
									Sid:       sid,
									ConfigNum: startNum,
								}
								//提交到raft
								kv.rf.Start(command)
								kv.logger.Printf("checkDelete: 分片ID=%d的允许删除命令已提交到Raft", sid)
								//虽然提交但是日志可能丢失，所以不能立即修改状态
								return
							} else if reply.Err == ErrDuplicate {
								kv.logger.Printf("checkDelete: 对方配置与当前配置相同，但已拉取分片，允许删除分片ID=%d", sid)
								//对方已经处理过删除了，只是自己没收到消息，所以可以检查状态，让自己进入下一个阶段。也不用加锁，直接提交命令，因为到时候会检查也是一样的
								//不过加个读锁可以提前检查一下
								command := ShardCommand{
									Type:      "handleDelete",
									Sid:       sid,
									ConfigNum: startNum,
								}
								//提交到raft
								kv.rf.Start(command)
								kv.logger.Printf("checkDelete: 分片ID=%d的允许删除命令已提交到Raft", sid)
								return
							}
						} else {
							kv.logger.Printf("checkDelete: 调用失败，分片ID=%d", sid)
						}
						//尝试下一个节点
						index = (index + 1) % len(addrs)
					}
				}(sid, gid, startNum, prevConfig.Groups[gid])
			}
			//先释放锁再等待，这样中间完成的分片可以被临时使用用于完成挑战2
			kv.mu.RUnlock()
			kv.logger.Printf("checkDelete: 等待所有删除检查协程完成，配置号=%d", startNum)
			//等待所有分片拉取结束，才可能进入下一个拉取循环，然后等待100ms，日志应当已经应用，若未应用说明可能丢失了
			wg.Wait()
			kv.logger.Printf("checkDelete: 本轮删除检查协程全部完成，配置号=%d", startNum)
		}
		time.Sleep(PullShardTimeout)
	}
}
